package com.clown.LoginFailDetect

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

// 输入的登录事件样例类
case class LoginEvent(userId: Long, ip: String, eventType: String, timestamp: Long)

// 输出报警信息样例类
case class LoginFailWarning(userId: Long, firstFailTime: Long, lastFailTime: Long, warningMsg: String)

object LoginFail {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val resource = getClass.getResource("/LoginLog.csv")
    val inputStream = env.readTextFile(resource.getPath)
    // 转换成样例类类型，并提取时间戳和watermark
    val loginEventStream = inputStream
      .map(data => {
        val arr = data.split(",")
        LoginEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(2)) {
        override def extractTimestamp(element: LoginEvent): Long = element.timestamp * 1000L
      })

    // 进行判断和检测，如果2秒之内连续登陆失败，输出报警信息
    val loginFailWarningStream = loginEventStream
      .keyBy(var2 => (var2.userId, var2.ip))
      .process(new LoginFailWarningResult(2, 3))

    loginFailWarningStream.print()
    env.execute("login fail warning job")
  }
}

class LoginFailWarningResult(failTimes: Int, second: Int) extends KeyedProcessFunction[(Long, String), LoginEvent, LoginFailWarning] {
  lazy val loginFailState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("loginFailState", classOf[LoginEvent]))
  lazy val timerTimestampState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timerTimestampState", classOf[Long]))

  override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[(Long, String), LoginEvent, LoginFailWarning]#Context, out: Collector[LoginFailWarning]): Unit = {
    // 判断当前登录事件是成功还是失败
    if (value.eventType.equals("fail")) {
      loginFailState.add(value)
      // 如果没有定时器，那么注册一个n秒后的定时器
      if (timerTimestampState.value() == 0) {
        val timestamp = value.timestamp * 1000 + second * 1000
        ctx.timerService().registerEventTimeTimer(timestamp)
        timerTimestampState.update(timestamp)
      }
    } else {
      // 如果是成功，那么直接清空状态和定时器，重新开始
      ctx.timerService().deleteEventTimeTimer(timerTimestampState.value())
      loginFailState.clear()
      timerTimestampState.clear()
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(Long, String), LoginEvent, LoginFailWarning]#OnTimerContext, out: Collector[LoginFailWarning]): Unit = {
    val allLoginFailList: ListBuffer[LoginEvent] = new ListBuffer[LoginEvent]()
    val iterator = loginFailState.get().iterator()
    while (iterator.hasNext) {
      allLoginFailList.+=(iterator.next())
    }
    // 判断登陆失败事件的个数，如果超过了上限，报警
    if (allLoginFailList.length >= failTimes) {
      out.collect(LoginFailWarning(ctx.getCurrentKey._1, allLoginFailList.head.timestamp, allLoginFailList.last.timestamp, "login fail in %ss for %s times.".format(second, allLoginFailList.length)))
    }
    // 清空状态
    loginFailState.clear()
    timerTimestampState.clear()
  }
}